# -*- coding: utf-8 -*-

# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: https://docs.scrapy.org/en/latest/topics/item-pipeline.html
import json

import pymysql
from kafka import KafkaProducer

from tutorial import settings
from tutorial.items import TrafficJamItem


class TutorialPipeline(object):
    def __init__(self):
        print("初始化与kafka的连接")
        self.producer = KafkaProducer(
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            bootstrap_servers=['192.168.88.101', '192.168.88.102', '192.168.88.103']
        )

    def process_item(self, item, spider):
        # connect = pymysql.connect(
        #     host=settings.MYSQL_HOST,
        #     port=settings.MYSQL_PORT,
        #     db=settings.MYSQL_DBNAME,
        #     passwd=settings.MYSQL_PASSWORD,
        #     user=settings.MYSQL_USER,
        #     charset='utf8',
        #     use_unicode=True,
        #     cursorclass=pymysql.cursors.DictCursor
        # )
        #
        # cursor = connect.cursor()
        # cursor.execute("""SELECT * from numberoftraffic where yearofnum=%s and place=%s""",
        #                (int(item["year"]), item["place"]))
        # tr = cursor.fetchone()
        # if tr:
        #     cursor.execute("""UPDATE numberoftraffic SET num=%s WHERE place=%s AND yearofnum=%s""",
        #                    (int(item["num"]), str(item["place"]), int(item["year"])))
        # else:
        #     cursor.execute("""INSERT INTO numberoftraffic(yearofnum,place,num) VALUES(%s,%s,%s)""",
        #                    (int(item["year"]), str(item["place"]), int(item["num"]))
        #                    )
        # connect.commit()
        if isinstance(item, TrafficJamItem):
            self.process_item_to_kafka(item)
        return item

    def process_item_to_kafka(self, item):
        print(item, "=======")
        tmp = {
            'year': item["year"],
            'place': item["place"],
            'number': item["num"]
        }
        self.producer.send("traffic_list", tmp)
        print(self)
        pass
